package org.shj.spark.sql

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.functions._

/**
 * Spark SQL中有很多内置函数可以用来对数据进行分析。 内置函数操作的结果 是返回一个Column对象，
 * 而DataFrame天生就是 "A distributed collection of data organized into named columns."
 * 这就为数据的复杂分析建立了坚实的基础，并提供了极大的方便性。例如： 我们在操作DataFrame的方法中可以随时调用内置函数 进行
 * 业务需要的处理。这之于我们构建业务逻辑而言，极大的减少不必须 的时间消耗。
 * 内置函数主要有以下几类：
 * 1. 聚合函数。 如： countDistinct、 sumDistinct
 * 2. 集合函数。 如： sort_array、 explode
 * 3. 日期、时间函数。 如： hour、quarter、 next_day
 * 4. 数学函数。 如： sin、atan、sqrt
 * 5. 开窗函数。 如： rowNumber
 * 6. 字符串函数。 如： concat， format_number
 * 7. 其它函数。 如： isNaN， sha、 callUDF
 * 
 */
object SparkSQLAgg {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("SparkSQLAgg").master("local").getOrCreate();
    spark.sparkContext.setLogLevel("WARN")
    
    import spark.sqlContext.implicits._
    
    val userData = Array("2016-03-12,yangmi,http://spark.apache.org", 
        "2016-03-12,yangmi,http://www.baidu.com",
        "2016-03-12,yangmi,http://www.163.com",
        "2016-03-13,zhaoliying,http://hive.apache.org",
        "2016-03-13,zhaoliying,http://hive.apache.org",
        "2016-03-13,yangmi,http://spark.apache.org",
        "2016-03-13,zhaoliying,http://www.163.com",
        "2016-03-13,zhuyin,http://hive.apache.org",
        "2016-03-14,zhuyin,http://spark.apache.org",
        "2016-03-14,zhaoliying,http://www.baidu.com"
        )
    
    val userRdd = spark.sparkContext.parallelize(userData)
    val userRowRDD = userRdd.map(row => {
      val arr = row.split(",")
      Row(arr(0), arr(1), arr(2))
    })
    
    val schema = StructType(Array(StructField("date", StringType, true),
                                  StructField("name", StringType),
                                  StructField("url", StringType)))
    val userDf = spark.createDataFrame(userRowRDD, schema)
    
    /**
     *  通过上面的  import spark.sqlContext.implicits._ ， 可以把形如 'columnName 东东隐式
     *  转换成 agg() 方法里需要 Column 类型的参数
     *  下面的语句的意思是： 把数据按 date 这个列分组，然后计算 name 这列去重后的总数
     */
    userDf.groupBy("date").agg('date, countDistinct('name)).show()
    
    //计算某天每个URL的访问量
    userDf.groupBy($"date", $"url").agg(count('name)).show()
    
    val empDf = spark.read.json("E:/workspace/scala/sparkjava/src/main/resources/employees.json")
    val deptDf = spark.read.json("E:/workspace/scala/sparkjava/src/main/resources/dept.json")
    
    empDf.filter("salary > 3000")
         .join(deptDf, empDf("deptCd") === deptDf("deptCd"))
         .groupBy(deptDf("deptCd"))
         .agg(avg(empDf("salary")), max(empDf("salary")))
         .show()
    
        
    spark.stop()
  }
}